package com.wenqiner.requesttool.core.thread;

import com.wenqiner.requesttool.domain.ParallelStatistics;
import com.wenqiner.requesttool.domain.RunnableInfo;
import com.wenqiner.requesttool.function.CustomTwoParamConsumer;
import com.wenqiner.requesttool.util.CommonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 并行线程工具
 *
 * @author rhf
 * @date 2023-03-23
 **/
public class ParallelThreadPoolTool {
    public static final Logger log = LoggerFactory.getLogger(ParallelThreadPoolTool.class);

    public static final Integer CORE_THREAD_SIZE = 1;
    public static final long KEEP_ALIVE_TIME = 60;
    public static final Integer DEFAULT_BLOCK_QUEUE_SIZE = 30;
    private boolean closeable = false;

    /**
     * 执行时间
     */
    private long requestFinishTimestamp;

    /**
     * 执行任务,返回成功数、失败数
     *
     * @param maxSize    最大并发数
     * @param consumer   执行方法
     * @param t          参数
     * @param runMinutes 执行时间
     * @param loopSize   循环次数
     * @return com.wenqiner.requesttool.domain.ParallelStatistics
     * @author rhf
     */
    public <T> ParallelStatistics parallelRun(Integer maxSize, CustomTwoParamConsumer<T, Integer> consumer, T t, Long runMinutes, Integer loopSize) {
        log.info("执行线程任务,并发线程数:[{}],运行时间:[{}], 最大循环次数:[{}]", maxSize, runMinutes, loopSize);
        long requestStartTimestamp = System.currentTimeMillis();
        requestFinishTimestamp = requestStartTimestamp + runMinutes * 1000;
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                maxSize
                , CORE_THREAD_SIZE + maxSize
                , KEEP_ALIVE_TIME
                , TimeUnit.SECONDS
                , new ArrayBlockingQueue<>(DEFAULT_BLOCK_QUEUE_SIZE)
                , new CustomerThreadFactory("Parallel"));
        closeable = false;
        List<CompletableFuture<RunnableInfo>> completableFutureList = new ArrayList<>();
        for (int i = 0; i < maxSize; i++) {
            final Integer index = i;
            CompletableFuture<RunnableInfo> completableFuture = CompletableFuture.supplyAsync(() -> {
                RunnableInfo runnableInfo = new RunnableInfo();
                int succ = 0;
                int total = 0;
                for (int loop = 0; loop < loopSize
                        && System.currentTimeMillis() <= requestFinishTimestamp
                        && !closeable; loop++) {
                    try {
                        consumer.accept(t, index);
                        succ++;
                    } catch (Exception e) {

                    } finally {
                        total++;
                    }
                }
                runnableInfo.setSuccessSize(succ);
                runnableInfo.setTotal(total);
                return runnableInfo;
            }, threadPoolExecutor);
            completableFutureList.add(completableFuture);
        }
        CompletableFuture<RunnableInfo>[] completableFutures = new CompletableFuture[completableFutureList.size()];
        AtomicReference<RunnableInfo> atomicReference = new AtomicReference<>();
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(completableFutureList.toArray(completableFutures))
                .whenComplete((x, e) -> {
                    RunnableInfo runnableInfo = completableFutureList
                            .stream()
                            .map(c ->
                                    c.getNow(new RunnableInfo()))
                            .reduce((v1, v2) -> {
                                v1.setTotal(v1.getTotal() + v2.getTotal());
                                v1.setSuccessSize(v1.getSuccessSize() + v2.getSuccessSize());
                                return v1;
                            }).get();
                    atomicReference.set(runnableInfo);
                });
        try {
            completableFuture.get();
        } catch (Exception e) {
            log.error("并发请求统计请求数据异常,异常原因 msg :[{}], e :", e.getMessage(), e);
        }
        long l = System.currentTimeMillis();
        if (l < requestFinishTimestamp) {
            requestFinishTimestamp = l;
        }
        log.info("关闭线程池、完成并发请求.......");
        //填充
        RunnableInfo runnableInfo = atomicReference.get();
        ParallelStatistics parallelStatistics = new ParallelStatistics();
        parallelStatistics.setFailureSize(Long.valueOf(runnableInfo.getTotal() - runnableInfo.getSuccessSize()));
        parallelStatistics.setSuccessSize(Long.valueOf(runnableInfo.getSuccessSize()));
        parallelStatistics.setTotal(Long.valueOf(runnableInfo.getTotal()));
        parallelStatistics.setStartTimestamp(requestStartTimestamp);
        parallelStatistics.setFinishTimestamp(requestFinishTimestamp);
        return parallelStatistics;
    }

    /**
     * 关闭线程池
     *
     * @author rhf
     */
    public void shutdown() {
        log.info("执行线程任务,手动停止线程运行...........");
        requestFinishTimestamp = System.currentTimeMillis();
        closeable = true;
    }


    public static void main(String[] args) {
        ParallelThreadPoolTool threadPoolUtil = new ParallelThreadPoolTool();
        String x = "hello";
        ParallelStatistics parallelStatistics = threadPoolUtil.parallelRun(3, ParallelThreadPoolTool::testPrint, x, 3L, Integer.MAX_VALUE);
        System.out.println("end:" + CommonUtil.formatJson(parallelStatistics));
    }

    public static void testPrint(String x, Integer index) {
        Thread t = Thread.currentThread();
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("threadName:" + t.getName() + "----index:" + index + "-----say:" + x);
    }
}
